import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
            //8.用java读取本地模拟文件封装成对象输出到控制台。（5分）
        DataStreamSource<String> file = env.readTextFile("data/test2.txt");
        file.print();
//9.用Java代码获取每个班级学生考试的总分打印到控制台。（8分）
        SingleOutputStreamOperator<Student> map = file.map(new MapFunction<String, Student>() {
            @Override
            public Student map(String value) throws Exception {
                String[] split = value.split(",");
                return new Student(split[0], split[1], Integer.parseInt(split[2]), split[3], split[4], Integer.parseInt(split[5]), split[6]);
            }
        });
        map.print();
        SingleOutputStreamOperator<Tuple2<String, Integer>> mapds = map.map(new MapFunction<Student, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(Student value) throws Exception {
                return new Tuple2<>("all", value.getSearch());
            }
        }).keyBy(0).sum(1);
        mapds.print();
//10.用Java代码获取每个班级学生考试的平均分打印到控制台。（8分）
        //11.用Java代码获取男生考试的分数打印到控制台。（8分）
        //12.用java代码获取张三这个老师一共有多少个学生。(8分)
        SingleOutputStreamOperator<Tuple2<String, Integer>> mapth = map.filter(x -> x.getTh().equals("张三")).map(new MapFunction<Student, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(Student value) throws Exception {
                return new Tuple2<>(value.getTh(), 1);
            }
        }).keyBy(0).sum(1);
        mapth.print();
        //13.用java代码获取超过24的有多少个人，请将结果过打印到控制台。(8分)
        SingleOutputStreamOperator<Tuple2<String, Integer>> mapcc = map.filter(x -> x.getSearch()>24).map(new MapFunction<Student, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(Student value) throws Exception {
                return new Tuple2<>(value.getName(), 1);
            }
        }).keyBy(0).sum(1);
        mapcc.print();
        env.execute();
    }
}
